package pres.man.interview.producerconsumer.signleton;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 生产者消费者问题
 * 使用ReentrantLock + Condition实现
 *
 * @author MAN
 * @date 2020-12-04 20:39
 * @project
 * @since
 */
public class LockAndCondition {
    private static final Lock LOCK = new ReentrantLock();
    private static final Condition empty = LOCK.newCondition();
    private static final Condition full = LOCK.newCondition();
    private static volatile AtomicInteger COUNT = new AtomicInteger(0);
    private static final ThreadPoolExecutor POOL = new ThreadPoolExecutor(5, 8, 20, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.CallerRunsPolicy());

    public static void main(String[] args) {
        LockAndCondition lockAndCondition = new LockAndCondition();
        for (int i = 0; i < 5; i++) {
            POOL.execute(() -> {
                for (; ; ) {
                    lockAndCondition.product();
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            POOL.execute(() -> {
                for (; ; ) {
                    lockAndCondition.consume();
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            POOL.execute(() -> {
                for (; ; ) {
                    lockAndCondition.consume();
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            POOL.execute(() -> {
                for (; ; ) {
                    lockAndCondition.consume();
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    private void product() {
        // 互斥访问仓库
        LOCK.lock();
        try {
            // 东西满了而阻塞
            while (COUNT.get() == 5) {
                System.out.println("仓库已满");
                full.await();
            }
            System.out.println("生产一个东西，此时有" + COUNT.incrementAndGet() + "个东西");
            // 通知被东西空了阻塞的线程
            empty.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            LOCK.unlock();
        }
    }

    private void consume() {
        // 互斥访问仓库
        LOCK.lock();
        try {
            // 东西空了而阻塞
            while (COUNT.get() == 0) {
                System.out.println("仓库已空");
                empty.await();
            }
            System.out.println("消费一个东西，此时有" + COUNT.decrementAndGet() + "个东西");
            // 通知被东西满了阻塞的线程
            full.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            LOCK.unlock();
        }
    }
}
